package com.zlsf.verticles;

import com.zlsf.conf.Config;
import com.zlsf.service.Jtt1078Decoder;
import com.zlsf.service.SessionManager;
import com.zlsf.util.ConstStr;
import com.zlsf.util.Packet;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 监听1078协议端口
 */
@Slf4j
public class MediaVerticle extends AbstractVerticle {

  byte[] block = new byte[4096];
  private final ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<String, Jtt1078Decoder> jtt1078DecoderMap = new ConcurrentHashMap<>();

  @Override
  public void start() throws Exception {
    super.start();
    int port = Config.getInstance().getConfig().getJsonObject("netWork").getInteger("port", 5556);
    this.createNetServer(port);
  }

  /**
   * 创建TCP服务器
   *
   * @param port
   */
  private void createNetServer(int port) {
    NetServerOptions options = new NetServerOptions();
    options.setRegisterWriteHandler(true);
    NetServer server = vertx.createNetServer(options);
    // 处理连接请求
    server.connectHandler(this::onConnect);
    server.listen(port).onSuccess(http -> log.info("tcp服务器启动成功，端口号：{}", port));
  }


  /**
   * 有连接上来
   *
   * @param socket
   * @return
   */
  private void onConnect(NetSocket socket) {
    String clientId = socket.writeHandlerID();
    socket.handler(buffer -> {
      List<Packet> packageList = this.decodePackage(buffer, clientId);
      for (Packet packet : packageList) {
        publish(packet, socket, vertx, clientId);
      }
    });

    // 监听客户端的退出连接
    socket.closeHandler(close -> {
      if (map.containsKey(clientId)) {
        String tag = map.get(clientId);
        map.remove(tag);
        SessionManager.remove(clientId, ConstStr.TAG);
        SessionManager.removeChannel(tag);
      }
    });
  }


  private List<Packet> decodePackage(Buffer inData, String clientId) {
    ByteBuf in = Unpooled.directBuffer(inData.length());
    in.writeBytes(inData.getBytes());

    int length = in.readableBytes();
    List<Packet> out = new ArrayList<>();
    Jtt1078Decoder decoder = this.jtt1078DecoderMap.computeIfAbsent(clientId, k -> new Jtt1078Decoder());
    for (int i = 0, k = (int) Math.ceil(length / 512f); i < k; i++) {
      int l = i < k - 1 ? 512 : length - (i * 512);
      in.readBytes(block, 0, l);

      decoder.write(block, 0, l);

      while (true) {
        Packet p = decoder.decode();
        if (p == null) break;
        out.add(p);
      }
    }
    in.release();
    return out;
  }

  /**
   * 解析h264裸流
   *
   * @param packet
   * @param socket
   * @param vertx
   * @param clientId
   */
  private void publish(Packet packet, NetSocket socket, Vertx vertx, String clientId) {
    packet.seek(8);
    String sim = packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD();
    int channel = packet.nextByte() & 0xff;
    String tag = sim + "-" + channel;
    map.put(clientId, tag);

    if (!SessionManager.contains(socket.writeHandlerID(), ConstStr.TAG)) {
      //PublishManager.getInstance().open(tag, vertx);
      SessionManager.set(socket.writeHandlerID(), ConstStr.TAG, tag);
      SessionManager.setChannel(socket, tag);
    }

    Integer sequence = SessionManager.get(socket.writeHandlerID(), "video-sequence");
    if (sequence == null) sequence = 0;
    // 1. 做好序号
    // 2. 音频需要转码后提供订阅
    int lengthOffset = 28;
    int dataType = (packet.seek(15).nextByte() >> 4) & 0x0f;
    int pkType = packet.seek(15).nextByte() & 0x0f;
    // 透传数据类型：0100，没有后面的时间以及Last I Frame Interval和Last Frame Interval字段
    if (dataType == 0x04) lengthOffset = 28 - 8 - 2 - 2;
    else if (dataType == 0x03) lengthOffset = 28 - 4;

    int pt = packet.seek(5).nextByte() & 0x7f;

    if (dataType == 0x00 || dataType == 0x01 || dataType == 0x02) {
      // 碰到结束标记时，序号+1
      if (pkType == 0 || pkType == 2) {
        sequence += 1;
        SessionManager.set(socket.writeHandlerID(), "video-sequence", sequence);
      }
      long timestamp = packet.seek(16).nextLong();
      byte[] data = packet.seek(lengthOffset + 2).nextBytes();
      this.publishData(tag, sequence, timestamp, pt, data, ConstStr.MEDIA_TYPE_VIDEO);
    } else if (dataType == 0x03) {
      long timestamp = packet.seek(16).nextLong();
      byte[] data = packet.seek(lengthOffset + 2).nextBytes();
      this.publishData(tag, sequence, timestamp, pt, data, ConstStr.MEDIA_TYPE_AUDIO);
    }
  }


  /**
   * 往集群发送裸流
   *
   * @param tag
   * @param sequence
   * @param timestamp
   * @param payloadType
   * @param data
   * @param mediaType
   */
  private void publishData(String tag, int sequence, long timestamp, int payloadType, byte[] data, String mediaType) {
    DeliveryOptions options = new DeliveryOptions();
    options.addHeader("tag", tag)
      .addHeader("sequence", sequence + "")
      .addHeader("timestamp", timestamp + "")
      .addHeader("payloadType", payloadType + "")
      .addHeader("mediaType", mediaType);
    vertx.eventBus().publish(tag + ConstStr.MEDIA_DATA_PUBLISH_ROUTER, data, options);
  }
}
